from kafka import KafkaConsumer
import json
import time
import requests
import re

def get_data():
    url="https://emcharts.dfcfw.com/suggest/stocksuggest2017.min.js"
    response=requests.get(url)
    result=re.compile('token:"(.*?)"').findall(response.text)
    print(result[0])

def get_topic_data():

    # 创建消费者实例
    consumer = KafkaConsumer(
        bootstrap_servers=['node1:9092', 'node2:9092', 'node3:9092', 'node4:9092'],  # 替换为你的Kafka服务器地址
        auto_offset_reset='earliest',  # 从最早的消息开始读取（'latest'表示仅读取新消息）
        enable_auto_commit=True,  # 自动提交偏移量
        group_id='test_group',  # 确保所有消费者实例使用相同的消费者组ID
        value_deserializer=lambda x: json.loads(x.decode('utf-8')) if x else None  # 反序列化消息值
    )

    # 订阅主题
    consumer.subscribe(['test'])  # 替换为你的主题名称

    print("Starting to poll for messages...")

    try:
        while True:
            # 轮询消息，设置超时时间为1秒（1000毫秒）
            msg_pack = consumer.poll(timeout_ms=1000, max_records=5)  # 一次最多拉取5条消息

            # 遍历每个分区的消息包
            for tp, messages in msg_pack.items():
                for message in messages:
                    print(f"Received message: {message.value} from topic {message.topic} at offset {message.offset}")

                    # 处理消息逻辑
                    # ...

                    # 提交偏移量，确保消息被正确处理
                    consumer.commit()
                    print("Offset committed.")

    except KeyboardInterrupt:
        print("Interrupted by user.")
    except Exception as e:
        print(f"An error occurred: {e}")
    finally:
        # 关闭消费者连接
        consumer.close()
        print("Consumer closed.")

if __name__ == '__main__':
    get_data()